package com.asinking.debezium;

import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import io.debezium.time.Timestamp;
import io.debezium.time.ZonedTimestamp;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;


public class MyDebeziumDeserializationSchema implements DebeziumDeserializationSchema<DebeziumObject> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<DebeziumObject> collector) throws Exception {
        Envelope.Operation op = Envelope.operationFor(sourceRecord);
        DebeziumObject debeziumObject = new DebeziumObject();
        // 是否快照
        Struct value = (Struct) sourceRecord.value();
        Struct source = value.getStruct("source");
        String isSnapshot = source.getString("snapshot");
        if (isSnapshot == null) {
            isSnapshot = "false";
        }
        debeziumObject.setIsSnapshot(isSnapshot);
        // 对DDL的数据进行处理
        if (value.schema().field("ddl") != null) {
            String ddlString = value.getString("ddl");
            // 只监控Alter Table的DDL语句
            if (ddlString.startsWith("ALTER TABLE")) {
                debeziumObject.setDDL(true);
                debeziumObject.setDdlString(ddlString);
                collector.collect(debeziumObject);
            }
            // 其他DDL 不做处理
            return;
        }
        // 获取主键
        Struct mysqlPk = (Struct) sourceRecord.key();
        List<Field> pkList = mysqlPk.schema().fields();
        HashMap<String, Object> mysqlPkMap = new HashMap<>();
        for (Field field : pkList) {
            String pkName = field.name();
            String pkValue = mysqlPk.get(pkName).toString();
            mysqlPkMap.put(pkName, pkValue);
        }
        debeziumObject.setPrikey(mysqlPkMap);

        // 判断类型
        if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
            debeziumObject.setRowKind("INSERT");
            collector.collect(extractAfter(debeziumObject, value));
        } else if (op == Envelope.Operation.DELETE) {
            debeziumObject.setRowKind("DELETE");
            collector.collect(extractBefore(debeziumObject, value));
        } else {
            debeziumObject.setRowKind("UPDATE");
            // 封装before数据
            debeziumObject.setOldValues(getBeforeData(debeziumObject, value, "before"));
            collector.collect(extractAfter(debeziumObject, value));
        }


    }

    public DebeziumObject extractAfter(DebeziumObject debeziumObject, Struct value) {
        return convert(debeziumObject, value, "after");
    }

    public DebeziumObject extractBefore(DebeziumObject debeziumObject, Struct value) {
        return convert(debeziumObject, value, "before");
    }

    public DebeziumObject convert(DebeziumObject debeziumObject, Struct value, String beforeOrAfter) {
        if (value.schema().field(beforeOrAfter) != null) {
            // 各字段键值对
            List<Field> allFieldsList = value.getStruct(beforeOrAfter).schema().fields();
            HashMap<String, Object> mysqlAllFieldsMap = new HashMap<>();
            for (Field field : allFieldsList) {
                String fieldName = field.name();
                // 对字段进行判空 防止NNP
                if (value.getStruct(beforeOrAfter).get(field) == null) {
                    continue;
                }
                // 处理debezium的时间问题
                if (Timestamp.SCHEMA_NAME.equals(field.schema().name())) {
                    Long l = value.getStruct(beforeOrAfter).getInt64(fieldName);
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                    mysqlAllFieldsMap.put(fieldName, formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(l), ZoneOffset.UTC)));
                    continue;
                }
                if (ZonedTimestamp.SCHEMA_NAME.equals(field.schema().name())) {
                    // 带时区
                    String hourPattern = "yyyy-MM-dd HH:mm:ss";
                    ZonedDateTime zdt = ZonedDateTime.parse(value.getStruct(beforeOrAfter).getString(fieldName));
                    LocalDateTime localDateTime = zdt.toLocalDateTime();
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern(hourPattern);
                    mysqlAllFieldsMap.put(fieldName, formatter.format(localDateTime.plusHours(8)));
                    continue;
                }
                Object fieldValue = value.getStruct(beforeOrAfter).get(field);
                mysqlAllFieldsMap.put(fieldName, fieldValue);
            }


            debeziumObject.setValues(mysqlAllFieldsMap);
        }
        return debeziumObject;
    }


    public HashMap<String, Object> getBeforeData(DebeziumObject debeziumObject, Struct value, String before) {
        if (value.schema().field(before) != null) {
            // 各字段键值对
            List<Field> allFieldsList = value.getStruct(before).schema().fields();
            HashMap<String, Object> mysqlAllFieldsMap = new HashMap<>();
            for (Field field : allFieldsList) {
                String fieldName = field.name();
                // 对字段进行判空 防止NNP
                if (value.getStruct(before).get(field) == null) {
                    continue;
                }
                // 处理debezium的时间问题
                if (Timestamp.SCHEMA_NAME.equals(field.schema().name())) {
                    Long l = value.getStruct(before).getInt64(fieldName);
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                    mysqlAllFieldsMap.put(fieldName, formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(l), ZoneOffset.UTC)));
                    continue;
                }
                if (ZonedTimestamp.SCHEMA_NAME.equals(field.schema().name())) {
                    // 带时区
                    String hourPattern = "yyyy-MM-dd HH:mm:ss";
                    ZonedDateTime zdt = ZonedDateTime.parse(value.getStruct(before).getString(fieldName));
                    LocalDateTime localDateTime = zdt.toLocalDateTime();
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern(hourPattern);
                    mysqlAllFieldsMap.put(fieldName, formatter.format(localDateTime.plusHours(8)));
                    continue;
                }
                Object fieldValue = value.getStruct(before).get(field);
                mysqlAllFieldsMap.put(fieldName, fieldValue);
            }

            return mysqlAllFieldsMap;

        }
        return null;
    }


    @Override
    public TypeInformation<DebeziumObject> getProducedType() {
        return TypeInformation.of(DebeziumObject.class);
    }
}

